package com.smyx.case2;

import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;

/**
 * zk 分布式锁
 */
@Slf4j
public class ZkDistributedLock {

    private static final String CONNECTION_STRING = "vm110:2181,vm111:2181,vm112:2181";
    private static final int SESSION_TIME = 2000;
    private static final String PARENT_NODE = "/locks";
    /** 当前节点 */
    private String currentNode = "";
    /** 上一个节点 */
    private String beforeNode = "";

    private final ZooKeeper zkClient;

    private final CountDownLatch currentLatch = new CountDownLatch(1);
    private final CountDownLatch waitLatch = new CountDownLatch(1);

    public ZkDistributedLock() throws IOException, InterruptedException, KeeperException {
        // 获取zk连接
        zkClient = new ZooKeeper(CONNECTION_STRING, SESSION_TIME, event -> {
            // currentLatch 如果连接上zk 可以释放
            if (event.getState() == Watcher.Event.KeeperState.SyncConnected) {
                currentLatch.countDown();
                log.info(">>>>>>>>>>> 连接上zk 可以释放");
            }
            // waitLatch 需要释放
            if (event.getType() == Watcher.Event.EventType.NodeDeleted && event.getPath().equals(beforeNode)) {
                waitLatch.countDown();
                log.info(">>>>>>>>>>> waitLatch 可以释放");
            }
        });
        // 等待zk正常连接后，往下走程序
        currentLatch.await();
        // 判断根节点/locks是否存在
        Stat stat = zkClient.exists(PARENT_NODE, false);
        if (stat == null) {
            log.info(">>>>>>>>>>> 根节点不存在，创建！！！");
            // 创建一下根节点 （永久节点）
            zkClient.create(PARENT_NODE, PARENT_NODE.replace("/", "").getBytes(),
                    ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
    }

    /**
     * 加锁
     */
    public void zkLock() throws InterruptedException, KeeperException {
        // 创建临时节点作为锁
        currentNode = zkClient.create(PARENT_NODE + "/seq-", null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
        // 测试效果
        Thread.sleep(10);
        // 判断创建的节点是否是最小的序号节点，如果是获取到锁；如果不是，监听他序号前一个节点
        List<String> children = zkClient.getChildren(PARENT_NODE, false);
        // 如果children 只有一个值，那就直接获取锁； 如果有多个节点，需要判断，谁最小
        if (children.size() == 1) {
            log.info(">>>>>>>>>>> 0 只有一个节点，直接获取锁！！！");
        } else {
            // 排序
            Collections.sort(children);
            // 获取节点名称 seq-00000000
            String thisNode = currentNode.substring((PARENT_NODE + "/").length());
            // 通过seq-00000000获取该节点在children集合的位置
            int index = children.indexOf(thisNode);
            // 判断
            if (index == -1) {
                log.info(">>>>>>>>>>> 数据异常");
            } else if (index == 0) {
                log.info(">>>>>>>>>>> 1 只有一个节点，直接获取锁！！！");
            } else {
                // 需要监听 他前一个节点变化
                beforeNode = "/locks/" + children.get(index - 1);
                zkClient.getData(beforeNode, true, new Stat());
                // 等待监听
                waitLatch.await();
            }
        }
    }

    /**
     * 解锁
     */
    public void zkUnLock() throws InterruptedException, KeeperException {
        zkClient.delete(currentNode,-1);
    }

}
